package algorithms;

import java.io.IOException;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * Each node will sum up the partial page rank scores it receives from its neighbors.
 * 
 */
public class PageRankReducer extends Reducer<Text, Text, Text, Text> {
        private static double DAMPENING_FACTOR = 0.5;
        
        public void reduce(Text key, Iterable<Text> values, Context context)
                        throws IOException, InterruptedException {              
        	
        		String neighbors = "";
        		double pageRank = 0.0;
        		for (Text value : values) {
        			String line = value.toString();
        			if (line.startsWith("VERTEX_STATE\t")) {
        				neighbors = line.substring("VERTEX_STATE\t".length());
        			} else {
        				pageRank += new Double(line);
        			}
        		}
        	
        		/*
                // Sum over all the values passed to the reducer.
                double summation = 0.0;
                
                StringBuilder neighbors = new StringBuilder("");
                for (Text value : values) {
                        String[] nodeAndPageRank = value.toString().split("\t");
                        summation += ((new Double(nodeAndPageRank[1])) / (new Double (nodeAndPageRank[2])));
                        
                        neighbors.append(nodeAndPageRank[0] + "\t");
                }
                
                // Calculate the new PageRank
                double pageRank = DAMPENING_FACTOR * (1 / (new Double(context.getConfiguration().get("total_nodes")))) +
                                (1 - DAMPENING_FACTOR) * summation;
                */
                context.write(key, new Text(pageRank + "\t" + neighbors.toString()));
        }
}
